<?php

namespace app\common;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\facade\Db;

class RabbitMQClient
{
    protected $connection;
    protected $channel;


    public function __construct()
    {
        // 从配置中读取 RabbitMQ 连接信息
        $config = config('rabbitmq')['rabbitmq'];

        // 创建连接
        $this->connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['user'],
            $config['password'],
            $config['vhost']
        );

        // 创建频道
        $this->channel = $this->connection->channel();
    }

    /**
     * 发送普通消息
     *
     * @param string $queueName 队列名称
     * @param mixed $message 消息内容
     * @return void
     */
    public function sendMessage($queueName, $message)
    {
        $result = [
            'status' => 'success',
            'message' => 'Message sent successfully.'
        ];

        try {
            // 参数验证
            if (!is_string($queueName) || empty($queueName)) {
                throw new \InvalidArgumentException("Queue name must be a non-empty string.");
            }

            if (empty($message)) {
                throw new \InvalidArgumentException("Message must not be empty.");
            }

            // 声明普通队列
            $this->channel->queue_declare($queueName, false, true, false, false);
            // 创建消息
            $msg = new AMQPMessage(
                $message,
                ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
            );
            // 发送消息到指定队列
            $this->channel->basic_publish($msg, '', $queueName);
        } catch (\InvalidArgumentException $e) {
            $result['status'] = 'error';
            $result['message'] = 'Invalid argument: ' . $e->getMessage();
        } catch (\Exception $e) {
            $result['status'] = 'error';
            $result['message'] = 'An error occurred: ' . $e->getMessage();
        }
        return $result;
    }


    /**
     * 接收普通消息
     *
     * @param string $queueName 队列名称
     * @param callable $callback 消息处理回调函数
     * @return void
     */
    public function receiveMessage($queueName, callable $callback)
    {
        $result = [
            'status' => 'success',
            'message' => 'All messages processed successfully.'
        ];

        try {
            // 参数验证
            if (!is_string($queueName) || empty($queueName)) {
                throw new \InvalidArgumentException("Queue name must be a non-empty string.");
            }

            if (!is_callable($callback)) {
                throw new \InvalidArgumentException("Callback must be a callable function.");
            }

            // 声明普通队列
            $this->channel->queue_declare($queueName, false, true, false, false);
            // 设置Qos，每次只消费一条
            $this->channel->basic_qos(null, 1, null);

            // 消费消息
            $consumerCallback = function ($msg) use ($callback, &$result) {
                try {
                    $sleepTime = Db::table('db_system_config')->where('key', 'SLEEP_TIME')->value('value') ?? 30; // 从数据库获取sleepTime，默认为30秒
                    // 调用业务回调处理消息
                    $isSuccessful = $callback($msg->body);
                    if ($isSuccessful) {
                        // 确认消息已处理
                        $msg->ack();
                    } else {
                        // 消息处理失败，将消息放在队列末尾
                        $this->channel->basic_publish($msg, '', $msg->get('routing_key'));
                        $msg->ack();
                        $result['status'] = 'partial_failure';
                        $result['message'] = 'Some messages failed to process and were requeued.';
                        sleep((int)$sleepTime); // 等待一段时间后重试
                    }
                } catch (\Exception $e) {
                    // 处理过程中发生异常，重新投递消息
                    $msg->reject(true); // 使用 basic_reject 代替 nack 进行测试，重新投递消息到队列
                    $result['status'] = 'partial_failure';
                    $result['message'] = 'Exception occurred while processing some messages: ' . $e->getMessage();
                }
            };

            // 消费消息
            $this->channel->basic_consume($queueName, '', false, false, false, false, $consumerCallback);

            // 保持消费者持续运行
            while ($this->channel->is_consuming()) {
                $this->channel->wait();
            }
        } catch (\InvalidArgumentException $e) {
            $result['status'] = 'error';
            $result['message'] = 'Invalid argument: ' . $e->getMessage();
        } catch (\Exception $e) {
            $result['status'] = 'error';
            $result['message'] = 'An error occurred: ' . $e->getMessage();
        } finally {
            $this->close();
        }
    }




    /**
     * 发送延时消息
     *
     * @param string $queueName 队列名称
     * @param int $delaySeconds 延迟时间（秒）
     * @param mixed $message 消息内容
     * @param string $dlxExchange 死信交换机名称
     * @param string $deadLetterQueue 死信队列名称
     * @return bool
     */
    public function sendDelayedMessage($queueName, $delaySeconds, $message, $dlxExchange = 'dlx_exchange', $deadLetterQueue = 'deadLetterQueue')
    {
        // 验证参数
        if (empty($queueName) || !is_numeric($delaySeconds) || $delaySeconds < 0 || empty($message)) {
            return [
                'status' => 'error',
                'message' => 'Invalid parameters: queueName and message must be non-empty, delaySeconds must be a non-negative number.'
            ];
        }

        // 初始化结果数组，默认消息发送成功
        $result = [
            'status' => 'success',
            'message' => 'Message sent successfully.'
        ];

        // 创建延时交换机和队列（如果未创建）
        $delayExchange = $queueName . '_delay_exchange'; // 延时交换机名称

        try {
            // 仅创建一次延时交换机
            $this->createDelayExchange($delayExchange);

            // 创建延时队列并绑定到延时交换机
            $delayQueue = $queueName;
            $this->createDelayQueue($delayQueue, $delaySeconds, $dlxExchange, $deadLetterQueue, $delayExchange);

            // 创建死信交换机（统一的死信交换机，供多个队列共享）
            $this->createDlxExchange($dlxExchange);

            // 创建并绑定普通队列到死信交换机
            $this->createProcessQueue($deadLetterQueue, $dlxExchange);

            // 创建消息并设置延迟时间（以毫秒为单位）
            $msg = new AMQPMessage($message, [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]);
            $msg->set('application_headers', new \PhpAmqpLib\Wire\AMQPTable());

            // 发送消息到延时交换机
            $this->channel->basic_publish($msg, $delayExchange);
        } catch (\Exception $e) {
            // 捕获任何异常并更新结果状态
            $result['status'] = 'error';
            $result['message'] = 'Failed to send message: ' . $e->getMessage();
        }
        return $result;
    }

    /**
     * 创建延时交换机（避免重复创建）
     *
     * @param string $exchangeName 交换机名称
     */
    private function createDelayExchange($exchangeName)
    {
        $this->channel->exchange_declare(
            $exchangeName, // 交换机名称
            'x-delayed-message', // 交换机类型，x-delayed-message 是延时交换机的类型
            false, // 是否持久化交换机，设置为 false 说明该交换机不会持久化，即RabbitMQ重启后交换机会丢失
            true, // 是否自动删除交换机，设置为 true 表示如果没有队列绑定到交换机，交换机会被自动删除
            false, // 是否持久化交换机的消息
            false, // 是否是排他交换机
            false, // 是否设置为内部交换机（无法被外部访问）
            new \PhpAmqpLib\Wire\AMQPTable(['x-delayed-type' => 'direct']) // 额外的参数，指定消息延迟类型为 direct（直接类型）
        );
    }


    /**
     * 创建延时队列，并绑定到延时交换机
     *
     * @param string $queueName 队列名称
     * @param int $delaySeconds 延迟时间（秒）
     * @param string $dlxExchange 死信交换机名称
     * @param string $processQueue 实际处理队列名称
     * @param string $delayExchange 延时交换机名称
     */
    private function createDelayQueue($queueName, $delaySeconds, $dlxExchange, $processQueue, $delayExchange)
    {
        $this->channel->queue_declare(
            $queueName,
            false,
            true,
            false,
            false,
            false,
            new \PhpAmqpLib\Wire\AMQPTable([
                'x-dead-letter-exchange' => $dlxExchange,
                'x-dead-letter-routing-key' => $processQueue,
                'x-message-ttl' => (int)($delaySeconds * 1000), // 设置消息TTL
            ])
        );
        $this->channel->queue_bind($queueName, $delayExchange);
    }

    /**
     * 创建死信交换机（避免重复创建）
     *
     * @param string $exchangeName 死信交换机名称
     */
    private function createDlxExchange($exchangeName)
    {
        $this->channel->exchange_declare($exchangeName, 'direct', false, true, false);
    }

    /**
     * 创建并绑定普通队列到死信交换机
     *
     * @param string $queueName 普通队列名称
     * @param string $dlxExchange 死信交换机名称
     */
    private function createProcessQueue($queueName, $dlxExchange)
    {
        $this->channel->queue_declare($queueName, false, true, false, false);
        $this->channel->queue_bind($queueName, $dlxExchange, $queueName);
    }



    /**
     * 发送延时消息
     *
     * @param string $queueName 队列名称
     * @param int $delaySeconds 延迟时间（秒）
     * @param mixed $message 消息内容
     * @return bool
     */
    public function sendDelayedMessageOld($queueName, $delaySeconds, $message)
    {
        // 验证参数
        if (empty($queueName) || !is_numeric($delaySeconds) || $delaySeconds < 0 || empty($message)) {
            return [
                'status' => 'error',
                'message' => 'Invalid parameters: queueName and message must be non-empty, delaySeconds must be a non-negative number.'
            ];
        }

        // 初始化结果数组，默认消息发送成功
        $result = [
            'status' => 'success',
            'message' => 'Message sent successfully.'
        ];

        // 创建延时交换机和队列（如果未创建）
        $delayExchange = $queueName . '_delay_exchange';
        $dlxExchange = $queueName . '_dlx_exchange';  // 死信交换机
        $processQueue = $queueName; // 实际处理的队列

        try {
            // 声明延时交换机
            $this->channel->exchange_declare(
                $delayExchange,
                'x-delayed-message',
                false,
                true,
                false,
                false,
                false,
                new \PhpAmqpLib\Wire\AMQPTable(['x-delayed-type' => 'direct'])
            );

            // 声明延时队列并绑定到延时交换机
            $delayQueue = $queueName . '_delay_queue';
            $this->channel->queue_declare(
                $delayQueue,
                false,
                true,
                false,
                false,
                false,
                new \PhpAmqpLib\Wire\AMQPTable([
                    'x-dead-letter-exchange' => $dlxExchange,
                    'x-dead-letter-routing-key' => $processQueue,
                    'x-message-ttl' => $delaySeconds * 1000 // 设置 TTL 以便消息开始自动转发
                ])
            );
            $this->channel->queue_bind($delayQueue, $delayExchange);

            // 声明死信交换机
            $this->channel->exchange_declare($dlxExchange, 'direct', false, true, false);

            // 声明处理队列并绑定到死信交换机
            $this->channel->queue_declare($processQueue, false, true, false, false);
            $this->channel->queue_bind($processQueue, $dlxExchange, $processQueue);

            // 创建消息并设置延迟时间（以毫秒为单位）
            $msg = new AMQPMessage($message, [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]);
            $msg->set('application_headers', new \PhpAmqpLib\Wire\AMQPTable(['x-delay' => $delaySeconds * 1000]));

            // 发送消息到延时交换机
            $this->channel->basic_publish($msg, $delayExchange);
        } catch (\Exception $e) {
            // 捕获任何异常并更新结果状态
            $result['status'] = 'error';
            $result['message'] = 'Failed to send message: ' . $e->getMessage();
            // 捕获任何异常并记录日志
        }
        return $result;
    }

    /**
     * 接收消息
     *
     * @param callable $callback 消息处理回调函数
     * @param string|null $queue 队列名称
     * @return void
     */
    public function receiveDelayedMessage(string $queueName, callable $callback)
    {
        // 验证参数
        if (empty($queueName) || !is_callable($callback)) {
            return [
                'status' => 'error',
                'message' => 'Invalid parameters: queueName must be a non-empty string and callback must be callable.'
            ];
        }

        // 初始化结果数组，默认所有消息处理成功
        $result = [
            'status' => 'success',
            'message' => 'All messages processed successfully.'
        ];

        try {
            // 死信交换机名称及实际处理队列名称
            $dlxExchange = $queueName . '_dlx_exchange';
            $processQueue = $queueName;

            // 声明死信交换机和处理队列
            $this->channel->exchange_declare($dlxExchange, 'direct', false, true, false);
            $this->channel->queue_declare($processQueue, false, true, false, false);

            // 将处理队列绑定到死信交换机，确保消息在延迟结束后能进入处理队列
            $this->channel->queue_bind($processQueue, $dlxExchange, $queueName);

            // 限制每次只处理一条消息，防止消息积压
            $this->channel->basic_qos(null, 1, null);

            // 定义消费者回调函数
            $consumerCallback = function ($msg) use ($callback, &$result) {
                try {
                    $sleepTime = Db::table('db_system_config')->where('key', 'SLEEP_TIME')->value('value') ?? 30; // 从数据库获取sleepTime，默认为30秒
                    // 调用业务逻辑处理消息
                    $isSuccessful = $callback($msg->body);

                    if ($isSuccessful) {
                        // 确认消息已成功处理
                        $msg->ack();
                    } else {
                        // 消息处理失败，将消息重新放入队列末尾
                        $this->channel->basic_publish($msg, '', $msg->get('routing_key'));
                        $msg->ack(); // 确认消息，以避免重复消费
                        // 更新结果状态，部分消息处理失败
                        $result['status'] = 'partial_failure';
                        $result['message'] = 'Some messages failed to process and were requeued.';
                        sleep((int)$sleepTime); // 等待一段时间后重试
                    }
                } catch (\Exception $e) {
                    // 处理过程中发生异常，拒绝消息并重新投递
                    $msg->reject(true); // 使用 basic_reject 重回队列
                    // 更新结果状态，部分消息处理失败并发生异常
                    $result['status'] = 'partial_failure';
                    $result['message'] = 'Exception occurred while processing some messages: ' . $e->getMessage();
                }
            };

            // 启动消息消费
            $this->channel->basic_consume($processQueue, '', false, false, false, false, $consumerCallback);

            // 持续监听队列，直到所有消息都被消费完毕
            while ($this->channel->is_consuming()) {
                $this->channel->wait();
            }
        } catch (\Exception $e) {
            // 捕获异常并输出错误信息
            $result['status'] = 'error';
            $result['message'] = 'An error occurred: ' . $e->getMessage();
        } finally {
            // 关闭连接
            $this->close();
        }
    }


    /**
     * 关闭连接和通道
     *
     * @return void
     */
    public function close()
    {
        if ($this->channel !== null) {
            $this->channel->close();
        }
        if ($this->connection !== null) {
            $this->connection->close();
        }
    }
}
